package com.elinshaw.hadoopdev.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * @author Administrator
 */
public final class HadoopUtils {

    public static FileSystem fileSystem = null;

    private HadoopUtils() {

    }

    public static void init() throws URISyntaxException, IOException, InterruptedException {
        Configuration configuration = new Configuration();
        configuration.set("fs.defaultFS", "hdfs://localhost:9000/");
        configuration.set("dfs.replication", "3");
        configuration.set("dfs.blocksize", "64M");
        fileSystem = FileSystem.get(new URI("hdfs://www.xiaoyilin.top:9000/"), configuration, "root");
    }

    /**
     * 创建文件
     *
     * @param path
     * @throws Exception
     */
    public static void create(String path) throws Exception {
        FSDataOutputStream output = fileSystem.create(new Path(path));
        output.write("hello hadoop".getBytes());
        output.flush();
        output.close();
    }

    /**
     * 判断是否存在
     *
     * @param path
     * @return
     * @throws Exception
     */
    public static boolean exists(String path) throws Exception {
        return fileSystem.exists(new Path(path));
    }

    /**
     * 查看文件内容
     */
    public static void cat(String path) throws Exception {
        FSDataInputStream in = fileSystem.open(new Path(path));
        IOUtils.copyBytes(in, System.out, 1024);
        in.close();
    }


    /**
     * 重命名
     *
     * @param oldPath
     * @param newPath
     * @throws Exception
     */
    public static void rename(String oldPath, String newPath) throws Exception {
        fileSystem.rename(new Path(oldPath), new Path(newPath));
    }

    /**
     * 上传文件
     *
     * @param localPath
     * @param hdfsPath
     * @throws Exception
     */
    public static void copyFromLocalFile(String localPath, String hdfsPath) throws Exception {
        fileSystem.copyFromLocalFile(new Path(localPath), new Path(hdfsPath));
    }

    /**
     * 上传文件
     */
    public static void copyFromLocalFileWithProgress() throws Exception {
        InputStream in = new BufferedInputStream(
                new FileInputStream(
                        new File("/Users/rocky/source/spark-1.6.1/spark-1.6.1-bin-2.6.0-cdh5.5.0.tgz")));
        FSDataOutputStream output = fileSystem.create(new Path("/hdfsapi/test/spark-1.6.1.tgz"),
                () -> {
                    System.out.print(".");  //带进度提醒信息
                });
        IOUtils.copyBytes(in, output, 4096);
    }


    /**
     * 下载文件
     */
    public static void copyToLocalFile(String localPath, String hdfsPath) throws Exception {
        fileSystem.copyToLocalFile(new Path(hdfsPath), new Path(localPath));
    }

    /**
     * 查看目录下所有文件
     *
     * @param path
     * @throws Exception
     */
    public static void listFiles(String path) throws Exception {
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path(path));
        for (FileStatus fileStatus : fileStatuses) {
            boolean isDir = fileStatus.isDirectory();
            short replication = fileStatus.getReplication();
            long len = fileStatus.getLen();
            System.out.println(fileStatus.getPath().toString());
        }
    }

    /**
     * 删除
     *
     * @param path
     * @throws Exception
     */
    public static void delete(String path) throws Exception {
        fileSystem.delete(new Path(path), true);
    }

    /**
     * 创建目录
     *
     * @param path
     * @return
     * @throws IOException
     */
    public static boolean mkdir(String path) throws IOException {
        return fileSystem.mkdirs(new Path(path));
    }


}
